package com.lagou.work5;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Properties;
/*
    flink 从kafka获取数据运行wordcount
    flink 并行度设置为2，占用2个slot
 */
public class FromKafka {
    public static void main(String[] args) throws Exception {
        //获取flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(2);
        //获取kafka配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","server1:9092");
        //flink消费指定topic
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>("topic3", new SimpleStringSchema(), properties);
        //从最早消息开始消费
        consumer.setStartFromEarliest();
        //获取数据
        DataStream<String> data = env.addSource(consumer);
        //对数据进行处理
        SingleOutputStreamOperator<Tuple2<String,Integer>> dataStream = data.flatMap(
                new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                        String[] s1 = s.split(" ");
                        for(String word : s1) {
                            collector.collect(new Tuple2<String, Integer>(word,1));
                        }
                    }
                }
        );
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = dataStream.keyBy(0).sum(1);
        //打印输出
        res.print();
        res.writeAsText("src/main/java/com/lagou/work5/wcout.txt");
        env.execute();

    }
}
